package rlib

import (
	"math/rand"
	"sync"
	"time"

	"github.com/golang/glog"
)

type BeatWaiter struct {
	rd *rand.Rand
	tl int
	tr int

	touched      bool
	masterName   string
	touchedMutex sync.Mutex
}

/**
confirm r - l <= 100
*/
func (bw *BeatWaiter) Init(l, r int, masterName string) {
	if l > r {
		t := r
		r = l
		l = t
	}
	bw.tl = l
	bw.tr = r
	bw.rd = rand.New(rand.NewSource(time.Now().Unix()))
	bw.touched = false
	bw.masterName = masterName
}

func (bw *BeatWaiter) Rrange(l, r int) int {
	t := r - l + 1
	lic := bw.rd.Intn(t)
	return lic + l
}

func (bw *BeatWaiter) GetMasterName() string {
	bw.touchedMutex.Lock()
	defer bw.touchedMutex.Unlock()
	return bw.masterName
}

func (bw *BeatWaiter) GetNewWaiter() (<-chan time.Time, error) {
	bw.touchedMutex.Lock()
	defer bw.touchedMutex.Unlock()
	if !bw.touched {
		return nil, ErrNoHeartBeatInPastTimeRange
	}
	bw.touched = false
	timex := bw.Rrange(bw.tl, bw.tr)
	return time.After(time.Duration(timex) * time.Millisecond), nil
}

// if false : do not pull log , else pull log
func (bw *BeatWaiter) Touch(masterName string) bool {
	bw.touchedMutex.Lock()
	defer bw.touchedMutex.Unlock()
	bw.touched = true
	if masterName != bw.masterName {
		bw.masterName = masterName
		return true
	}
	return false
}

// -------------------------------------------------

type RaftCore struct {
	Conf NodeConfig

	Term       uint64
	State      string
	LogId      uint64
	MasterName string

	VoteLock sync.Mutex

	fsm *FileStorageManager

	beatWaiter *BeatWaiter

	// loop controller
	termLoopControl   chan bool
	masterLoopControl chan bool
}

func (rc *RaftCore) Init(startCh chan int) {
	rc.fsm = &FileStorageManager{}
	rc.beatWaiter = &BeatWaiter{}
	rc.beatWaiter.Init(3000, 5000, "")
	rc.termLoopControl = make(chan bool, 1)
	rc.masterLoopControl = make(chan bool, 1)
	rc.Term = 0
	rc.State = "follower"
	rc.Conf.NodeName = ClusterConfigSingleton.NodeName
	go func() {
		// wait for the start ch
		<-startCh
		glog.Infof("raftserver %s is beginning", ClusterConfigSingleton.NodeName)
		go rc.followerLoop()
	}()
}

func (rc *RaftCore) CallForVote(req *CallForVoteReq, rsp *CallForVoteRsp) {
	if rc.Term >= req.Term || rc.LogId > req.LogId {
		rsp.ErrMsg = "NO"
		return
	}
	rc.VoteLock.Lock()
	rc.State = "follower"
	rc.Term = req.Term
	rc.MasterName = req.NodeName
	rsp.ErrMsg = "OK"
	rc.VoteLock.Unlock()
}

func (rc *RaftCore) PullLog(req *PullLogReq, rsp *PullLogRsp) {
	rsp.LogItems = make([]string, 0)
	rc.fsm.PeekFromLogId(req.LogId, &rsp.LogItems)
}

func (rc *RaftCore) Beat(req *BeatReq, rsp *BeatRsp) {
	rsp.ErrMsg = "OK"
	lastMaster := rc.beatWaiter.GetMasterName()
	resetMaster := rc.beatWaiter.Touch(req.MasterName)
	if resetMaster {
		glog.Infof("reset master, from M1:%s to M2:%s", lastMaster, req.MasterName)
	} else {
		glog.Info("do not reset master")
	}
}

func (rc *RaftCore) Register(req *RegisterReq, rsp *RegisterRsp) {
	addNodeSucc := ClusterConfigSingleton.AddNewNode(NodeConfig{
		NodeName: req.Name,
		Addr:     req.Addr,
	})
	if !addNodeSucc {
		rsp.ErrMsg = ErrAddNodeFailed.Error()
	} else {
		rsp.ErrMsg = "OK"
	}
	// after register boardcast
	rc.masterHeartBeat()
}

func (rc *RaftCore) followerLoop() {
	beMaster := false
followerBegin:
	for {
		nowTimer, err := rc.beatWaiter.GetNewWaiter()
		if err != nil {
			glog.Error(err)
			rc.State = "candidate"
			beMaster = rc.candidateTask()
			goto followerEnd
		} else {
			<-nowTimer
		}
	}
followerEnd:
	if !beMaster {
		goto followerBegin
	}
	rc.masterLoop()
}

func (rc *RaftCore) masterLoop() {
	glog.Info("begin master loop")
	rc.State = "master"
	for {
		select {
		case <-rc.masterLoopControl:
			{
				goto masterLoopEnd
			}
		case <-time.After(1000 * time.Millisecond):
			{
				rc.masterHeartBeat()
			}
		}
	}
masterLoopEnd:
}

func (rc *RaftCore) masterHeartBeat() {
	glog.Info("begin to heartbeat")
	stbc := &SingleTypeBatchClient{}
	stbc.Init()
	ClusterConfigSingleton.rdlock.Lock()
	for k := range ClusterConfigSingleton.Nodes {
		if k != rc.Conf.NodeName {
			apiClient := &ApiClient{}
			apiClient.Init(k)
			stbc.Append(apiClient, &BeatRsp{})
		}
	}
	ClusterConfigSingleton.rdlock.Unlock()
	glog.Infof("now begin to heartbeat, followers size=%d", stbc.Size())
	stbc.BeginToRun(&BeatReq{MasterName: rc.Conf.NodeName})
}

func (rc *RaftCore) candidateTask() bool {
	stbc := &SingleTypeBatchClient{}
	stbc.Init()
	for k := range ClusterConfigSingleton.Nodes {
		if k != rc.Conf.NodeName {
			ApiClient := &ApiClient{}
			ApiClient.Init(k)
			rc.Term++
			stbc.Append(ApiClient, &CallForVoteRsp{})
		}
		stbc.BeginToRun(&CallForVoteReq{
			Term:     rc.Term,
			LogId:    rc.LogId,
			NodeName: rc.Conf.NodeName,
		})
	}
	beMaster := stbc.WaitAfterCount(len(ClusterConfigSingleton.Nodes) / 2)
	return beMaster
}
